package com.surfilter.massdata.spark.task.daystat2;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.SAN2201;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;

/**
 * 关注网站访问量地市分布
 * 本省网民访问重点网站的访问量运营商占比情况（本省网民是指CIP在BR2002中存在，省内网站是指DIP在BR2002中存在）
 */
public class WebSiteVisitsCityTask_New implements ISparkTask{
	
	private static final Log log = LogFactory.getLog(WebSiteVisitsCityTask_New.class);
	private static final long serialVersionUID = 1L;
	private String outName;

	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,Map<String, String> commandMap) {
		long start=System.currentTimeMillis();
		log.info("==============WebSiteVisitsCityTask begin====================");
		List<OutQueueEntity> outList = new ArrayList<OutQueueEntity>();
		try{
			DataFrame dataSource = dataFrames.get("dws_hour_san2201");
			
			String dayStr = commandMap.get("-d");
			Date date = DateUtil.getExecDate(DateUtil.getYesterday(),dayStr);
			
			//JavaRDD<Map<String, Object>> domainRDD = getCipDipFilterRDD(dataSource);
	        //JavaRDD<Map<String, Object>> rdd = getRdd(domainRDD);
	        
			DataFrame result = createRdd(date, dataSource);
			outList.add(new OutQueueEntity(outName, result));
			
			CommonUtils.deleteTaskTable("SAN2201", date, "WEBSITE_FOCUS_AREA", 0,"day");
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		long end=System.currentTimeMillis();
		double min=(end-start)*1.0/(1000*60);
		System.out.println("WebSiteVisitsCityTask_New:exectime: "+min+" min............");
		return outList;
	}

	@SuppressWarnings({"serial" })
	private DataFrame createRdd(final Date date,DataFrame dataSource) throws InterruptedException {
		
		String sql = "select sum(count_value)as count,BUSS_VALUE,TOP_DOMAIN from dws_hour_san2201_temp where BUSS_TYPE='WEBSITE_FOCUS_AREA' and ssfl=0 GROUP BY BUSS_VALUE,TOP_DOMAIN";
		DataFrame df = dataSource.sqlContext().sql(sql);
		JavaRDD<SAN2201> javaRDD = df.toJavaRDD().map(new Function<Row, Map<String,String>>() {

			@Override
			public Map<String, String> call(Row row) throws Exception {
				Map<String, String> map = new HashMap<String, String>();
				map.put("count", row.getAs("count").toString());
				map.put("top_domain", row.getAs("TOP_DOMAIN").toString());
				map.put("city", row.getAs("BUSS_VALUE").toString());
				return map;
			}
		}).mapPartitions(new FlatMapFunction<Iterator<Map<String,String>>, SAN2201>() {

			@Override
			public Iterable<SAN2201> call(Iterator<Map<String, String>> it) throws Exception {
				List<SAN2201> list = new ArrayList<SAN2201>();
				try{
					while(it.hasNext()){
						Map<String, String> map = it.next();
						String domain = (String) map.get("top_domain");
						if(StringUtils.isNotBlank(domain)){
		        			SAN2201 sa = new SAN2201();
		            		sa.setYear(DateUtil.getCurrentYear(date));
		            		sa.setHalf_year(DateUtil.getHalfYear(date));
		            		sa.setQuarter(DateUtil.getQuarter(date));
		            		sa.setMonth(DateUtil.getCurrentMonth(date));
		            		sa.setWeek(DateUtil.getCurrentWeek(date));
		            		sa.setDay(DateUtil.getCurrentDay(date));
		            		sa.setHour(0);
		            		sa.setSsfl(0);
		            		sa.setBuss_type("WEBSITE_FOCUS_AREA");
		            		sa.setBuss_value(CommonUtils.valueOf(map.get("city")));
		            		sa.setTop_domain(domain);
		            		sa.setCount_value(Long.parseLong(CommonUtils.valueOf(map.get("count"))));
		            		list.add(sa);
		        		}
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return list;
			}
		});
		DataFrame resultDf = dataSource.sqlContext().createDataFrame(javaRDD, SAN2201.class);
		return resultDf;
	}
}
